package com.example.biz.order.el.mq;

import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * 消息消费
 *
 * @author lr
 * @create 2021/9/16
 */
@Slf4j
@Component
@RequiredArgsConstructor
@RabbitListener(queues = "")
public class ReceiverToProjectCost {

    @Resource
    private RedisTemplate redisTemplate;

    @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = "el_admin", type = ExchangeTypes.DIRECT),
            key = "el_admin_message",
            value = @Queue(value = "el_admin_queue", autoDelete = "false"),
            ignoreDeclarationExceptions = "true"),
            concurrency = "1",
            ackMode = "MANUAL"
    )
    public void acceptProcess(String json, Channel channel, Message messages, CorrelationData correlationData) throws Exception {
        log.info("进入mq消费方法=======================" + correlationData);
        if (null == json) {
            return;
        }
        try {
            this.process(json, channel, messages);
        } catch (Exception e) {
            log.error("发送造价项目消息错误：{}", e);
        }
    }

    public void process(String json, Channel channel, Message message) {
        String messageId = (String) message.getMessageProperties().getHeaders().get("message_id");
        log.info("消息id唯一标识为：{}", messageId);
        //不存在设置成功，消息没有消费,存在说明已经进行过消费
        final Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(messageId, "0", 10, TimeUnit.SECONDS);
        if (aBoolean) {
            redisTemplate.opsForValue().setIfAbsent(messageId, "1", 10, TimeUnit.SECONDS);
            try {
                // 是否批量. true：将一次性ack所有小于deliveryTag的消息。
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                e.printStackTrace();
                log.error("错误：{}", e);
            }
        } else {
            if ("1".equals(redisTemplate.opsForValue().get(messageId))) {
                try {
                    // 是否批量. true：将一次性ack所有小于deliveryTag的消息。
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("错误：{}", e);
                }
            }
        }

    }

}